

import java.io.File;
import java.io.IOException;
import java.util.Calendar;
import java.util.HashSet;
import java.util.Map;
import java.util.HashMap;
import java.util.Set;

import com.parabon.client.*;
import com.parabon.io.InvalidLoginException;

public class NativeApp implements TaskResultListener, TaskExceptionListener {
  public static final String JOB_TYPE_ATTRIBUTE         = "JobType";
  public static final String JOB_NAME_ATTRIBUTE         = "JobName";
   
  public static final String INPUT_VALUE_ATTRIBUTE = "InputValue";
  public static final String path = "C:\\Documents and Settings\\chandana\\Desktop\\datafiles";
  public static final String JOB_TYPE = "NativeApp";
  public static final String DEFAULT_JOB_NAME = "DefaultNativeApp";
  public static final String JOB_INPUT_VALUES_ATTRIBUTE = "InputValues";
  private Set<String> pendingTasks = new HashSet<String>();
  
  public static final String EXECUTABLE_ELEMENT_FILENAME = "nativeTask.jar";
  public String nativeBundleName = "square.jar"; 


  private String jobName;
  private SessionManager manager;
  private Job job;
  private boolean taskComplete = false;

  private NativeApp(String jobName) {
    this.jobName = jobName;
    createSession();
  }
  
  private void createSession() {
    System.out.println("Connecting to session manager");
    try {
      manager = SessionManager.getInstance();
      manager.reestablish();
      System.out.println("Connected via " + manager);
    }
    catch (ServerCommunicationException e) {
      System.err.println("Unable to establish connection to server");
      System.err.println(e);
      System.exit(1);
    }
    catch (InvalidLoginException e) {
      System.err.println("Login failed");
      System.err.println(e);
      System.exit(1);
    }
  }
  
  /**
   * Closes the connection to the server
   */
  public void closeSession() {
    if (manager != null) {
      System.out.println("Disconnecting from server");
      manager.destroy();
      manager = null;
      System.out.println("Disconnected");
    }
  }

  /**
   * Finds the job in the current session
   */
  public boolean findJob() {
    assert(job == null);
    
    for (Job currJob : manager.getJobs()) {
      String currJobType = currJob.getAttributes().get(JOB_TYPE_ATTRIBUTE);
      String currJobName = currJob.getAttributes().get(JOB_NAME_ATTRIBUTE);
      if ((currJobType != null) && currJobType.equals(JOB_TYPE) &&
          (currJobName != null) && currJobName.equals(jobName)) {
        job = currJob;
        return true;
      }
    }

    return false;
  }

  public void launch(String[] inputValues,boolean addListener) throws IOException 

{
    assert(job == null);
    
    System.out.println("Creating " + JOB_TYPE + " with name " + jobName);
    StringBuilder inputValuesAttribute = new StringBuilder();
    for (String inputValue : inputValues) {
      if (inputValuesAttribute.length() != 0) {
        inputValuesAttribute.append(' ');
      }
      inputValuesAttribute.append(inputValue);
    }
   
    Map<String, String> jobAttributes = new HashMap<String, String>();
    jobAttributes.put(JOB_TYPE_ATTRIBUTE, JOB_TYPE);
    jobAttributes.put(JOB_NAME_ATTRIBUTE, jobName);
    jobAttributes.put(JOB_INPUT_VALUES_ATTRIBUTE,
            inputValuesAttribute.toString());
    
    job = manager.createJob(jobAttributes);
    
    File demoDirectory = new File(System.getProperty("demo.home", "."));
    File taskJarFile = new File(demoDirectory, EXECUTABLE_ELEMENT_FILENAME);
    System.out.println("native file exists1");
    String taskJarID = null;
    try {
      taskJarID = job.addJavaExecutableElement(taskJarFile);
    }
    catch (IOException e) {
      System.err.println("Can't access task jar file at " + taskJarFile);
      System.exit(0);
    }
    for (String inputValue : inputValues) {
        pendingTasks.add(inputValue);
      }

    // Add native data element containing square executable to job
    File nativeBundleFile = new File(demoDirectory + File.separator + nativeBundleName);
    System.out.println("native file exists2");
    String  dataElementID  = job.addDataElement( nativeBundleFile, ElementPacking.PARABON_NATIVE_JAR );
    
    
    if (addListener) {
      addListener();
    }
    
    //
    // Start the task
    //
    SerializableTaskSpec taskSpec = new SerializableTaskSpec();
    taskSpec.addRequiredExecutableElement(taskJarID);
    DataElementProxy dataElementProxy = new DataElementProxy(dataElementID);
    for (String inputValue : inputValues) {
         Map<String, String> taskAttributes = new HashMap<String, String>();
         taskAttributes.put(INPUT_VALUE_ATTRIBUTE, inputValue);
         System.out.println(path + File.separator + inputValue);
         File dataFile = new File(path + File.separator + inputValue);
         String  dataID  = job.addDataElement( dataFile);
         System.out.println("data file exists1");
         DataElementProxy dataProxy = new DataElementProxy(dataID);
         
         taskSpec.setTask(new NativeTask(dataElementProxy, "frontier",dataProxy) );
         TaskProxy task = job.addTask(taskSpec, taskAttributes);
         task.start();
         Calendar cal=Calendar.getInstance();
         String date=(cal.getTime()).toString();
         System.out.println("Created task at time:"+date);
      }
    
   System.out.println("ALL TASKS STARTED");
  }

  public void addListener() {
    job.addListener(this);
  }
  
  public synchronized void waitUntilComplete() {
    System.out.println("Waiting for results...");
    while (!pendingTasks.isEmpty()) {
      try {
        wait();
      }
      catch (InterruptedException e) {}
    }
  }
    
  /**
   * Destroys the job and all tasks in the job
   */
  public void remove() {
    if (job != null) {
      job.remove();
      job = null;
    }
  }

  private String getInputValue(TaskEvent event) {
	  System.out.println("into getInputValue()");
	  String i=(String)event.getTaskAttributes().get(INPUT_VALUE_ATTRIBUTE);
	  return i;
  }
  
  private synchronized void taskComplete(String inputValue) {
	  pendingTasks.remove(inputValue);
	    if (pendingTasks.isEmpty()) {
	      notifyAll();
	    }
	  }
  public void resultsPosted(TaskResultEvent event) {
	  String inputValue = getInputValue(event);
      NativeTask.Result result = (NativeTask.Result)event.getResultsObject();
      System.out.println("The square is " + result.getSquare());
      System.out.println("The hostname is " + result.getHostname());
      System.out.println("Time on the compute engines is" + result.getEntertime());
      Calendar cal=Calendar.getInstance();
      String date=(cal.getTime()).toString();
      System.out.println("results posted at time:"+date);
      taskComplete(inputValue);
  }

  public void exceptionThrown(TaskExceptionEvent event) {
	   String inputValue = getInputValue(event);
    System.out.println(
      "Exception (" + event.getCode() + ")" + " while running task");
    System.out.println(event.getDescription());
    System.out.println();

    taskComplete(inputValue);
  }
  
  public static void usage() {
    System.err.println("Usage: " + NativeApp.class + " (launch|listen|remove)");
  }

  public static void main(String[] args) throws IOException {
    boolean launch = false;
    boolean listen = false;
    boolean remove = false;
    
    if (args.length == 0) {
      usage();
      System.exit(1);
    }
    for (String arg : args) {
      if (arg.equals("launch")) {
        launch = true;
      } else if (arg.equals("listen")) {
        listen = true;
      } else if (arg.equals("remove")) {
        remove = true;
      } else {
        usage();
        System.exit(1);
      }
    }
    
    String jobName = System.getProperty("jobName", DEFAULT_JOB_NAME);
    NativeApp dataApp = new NativeApp(jobName);
    
    if (launch) {
      if (dataApp.findJob()) {
        System.err.println(
          "Job of type " + JOB_TYPE + " with name " + jobName +
          " already exists.");
        System.err.println("Use the \"remove\" mode to remove the job.");
        System.exit(1);
      }
      String[] inputValues = new String[] {"datafile1.txt","datafile2.txt","datafile3.txt","datafile4.txt"};
      
      dataApp.launch(inputValues,listen);
    } else {
      if (!dataApp.findJob()) {
        System.err.println(
          "Job of type " + JOB_TYPE + " with name " + jobName +
          " does not exist.");
        System.err.println("Use the \"launch\" mode to create a new job.");
        System.exit(1);
      }
    }
    
    if (listen) {
      if (!launch) {
        dataApp.addListener();
      }
      dataApp.waitUntilComplete();
    }
    
    if (remove) {
      dataApp.remove();
    }
     System.out.println("above closeSession()");
    dataApp.closeSession();
  }
}
